package com.kamistoat.rocketmqlearning.orderMsg;

import com.kamistoat.rocketmqlearning.common.R;
import com.kamistoat.rocketmqlearning.msgProducer.MsgProducerFactory;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@RequestMapping("/orderMsg")
@Controller
public class OrderMsgController {
    @Autowired
    @Qualifier("msgProducer")
    DefaultMQProducer msgProducer;

    @Autowired
    MsgProducerFactory msgProducerFactory;

    @RequestMapping("sync")
    @ResponseBody
    public R syncProducer() {
        try {
            // DefaultMQProducer msgProducer = msgProducerFactory.newMsgProducer();
            // msgProducer.start();
            /**
             * RocketMQ官方建议，使用业务中的最小细粒作为'部分有序'的Queue选择依据。
             * 例如用户Id或订单Id等等，这样可以实现同一终端用户的消息按照顺序处理，不同用户的消息则不需要保证顺序
             * 下面就是模拟在生产环境下，同时为5个用户提供服务，他们都是要发送4条消息。
             * 实际生产环境中这种操作显然是多线程调用同一个方法。这里使用异步进行模拟，也能更明确看出线程关系
             */
            CompletableFuture<Void> future_1 = CompletableFuture.runAsync(() -> {
                asyncUserSimulate(1, this.msgProducer);
            });
            CompletableFuture<Void> future_2 = CompletableFuture.runAsync(() -> {
                asyncUserSimulate(2, this.msgProducer);
            });
            CompletableFuture<Void> future_3 = CompletableFuture.runAsync(() -> {
                asyncUserSimulate(3, this.msgProducer);
            });
            CompletableFuture<Void> future_4 = CompletableFuture.runAsync(() -> {
                asyncUserSimulate(4, this.msgProducer);
            });
            CompletableFuture<Void> future_5 = CompletableFuture.runAsync(() -> {
                asyncUserSimulate(5, this.msgProducer);
            });
            CompletableFuture.allOf(future_1, future_2, future_3, future_4, future_5).get();
            return R.ok();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return R.error(e.getMessage());
        }finally {
            // msgProducer.shutdown();
        }
    }

    /**
     * 通过userId实现'部分顺序'
     */
    private void asyncUserSimulate(int userId, DefaultMQProducer msgProducer) {
        System.out.println("用户" + userId + "使用线程:" + Thread.currentThread());
        try {
            // 使用生产者发送10条消息
            for (int i = 0; i < 4; i++) {
                Message mQMsg = new Message();
                mQMsg.setTopic("orderTopic");
                byte[] body = ("Hi " + userId + ":" + i).getBytes(StandardCharsets.UTF_8);
                mQMsg.setBody(body);
                /**
                 * 一种向队列选择器传递参数的方法是，将参数包含在消息的 keys 中
                 */
                mQMsg.setKeys(String.valueOf(userId));

                /**
                 * 使用指定队列选择器算法的send方法发送消息。
                 * @param mQMsg 要发送的消息，该参数将作为下方覆写select函数中的第二个参数传入
                 * @param MessageQueueSelector 队列选择器接口
                 * @param null 该参数将作为下方覆写select函数中的第三个参数传入.如果不需要则写null即可
                 */
                SendResult sendResult = msgProducer.send(mQMsg, new MessageQueueSelector() {
                    /**
                     * 为队列选择器接口覆写的select选择算法
                     * @param mqs 由Broker返回，队列列表
                     * @param msg 与mQMsg相同
                     * @param arg 将自动填入 send() 方法的第三个参数
                     * @return 返回mqs列表中的一个队列
                     */
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        /**
                         * 直接用 'userId.hashcode()%队列数量' 定位Queue。
                         * 显然由于散列冲突，有可能不同userId的消息投到相同Queue，但显然不重要，因为我需求的就是相同userId的消息是顺序的
                         * 参数可以在消息的keys中进行传递，也可以通过send()的第三个参数进行传递
                         */
                        int index = Integer.valueOf(msg.getKeys()).hashCode() % mqs.size();
                        // int index = ((Integer) arg).hashCode() % mqs.size();
                        return mqs.get(index);
                    }
                }, userId);
                System.out.println("用户" + userId + ":" + sendResult);
            }
        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
